/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.store.hbase.hbaseop;

import cn.ac.iie.ulss.indexer.runenvs.GlobalParas;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.util.Utf8;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;


/**
 *
 * @author work
 */
public class HbaseUtilOp {

    public static Logger log = Logger.getLogger(HbaseUtilOp.class.getName());
    private static HConnection connection = null;
    private static Configuration conf = null;

    public static boolean initHbaseConnetion() {
        conf = new Configuration();
        try {            
            conf.set("hbase.zookeeper.quorum", GlobalParas.hbaseZk);
            conf.set("hbase.zookeeper.property.clientPort", GlobalParas.hbasePort+"");
            conf.set("hbase.master", GlobalParas.hbaseMaster);          
            //conf.addResource("xml/core-site.xml");
            //conf.addResource("xml/hdfs-site.xml");
            //conf.addResource("xml/mapred-site.xml");
            //conf.addResource("xml/yarn-site.xml");
            connection = HConnectionManager.createConnection(conf);
            
            log.info("the hbase conf is:"+conf);
            GlobalParas.conf = conf;
            //HbaseUtilOp.connection = HConnectionManager.createConnection(HbaseUtilOp.conf);
            GlobalParas.connection = connection;
            log.info("init connection to hbase ok ... ");
           
            return true;
        } catch (Exception e) {
            log.info("init the connection failed:" + e);
            return false;
        }
    }

    public static HTable getDefaultHTable(String zkUrls, int port, String tableName, String master) {
        //initHbaseConnetion();
        if (GlobalParas.connection == null) {
            while (!HbaseUtilOp.initHbaseConnetion()) {
                log.info("retry init hbase connection to " + zkUrls);
                try {
                    Thread.sleep(10000);
                    continue;
                } catch (Exception ex) {
                    log.error(ex, ex);
                }
            }
        }
        log.info("will get table from " + zkUrls);
        HTable table = null;
        try { 
            
            table = new HTable(GlobalParas.conf, tableName);           
            table.setAutoFlush(false);
            table.setOperationTimeout(15000);
            table.setWriteBufferSize(1024 * 1024 * 2);
        } catch (Exception e) {
            log.error(e, e);
        }
        return table;
    }
    public static boolean getByRowkey(String tableName,String rowkey){
        //initHbaseConnetion();
        HTable table = null;        
        try {
            if (GlobalParas.connection == null) {                
                initHbaseConnetion();
            }           
            table = new HTable(GlobalParas.conf, Bytes.toBytes(tableName));           
            Get get = new Get(Bytes.toBytes(rowkey));
            Result r = table.get(get);            
            return !r.isEmpty();
        }catch(Exception e){           
            return false;
        }                
    }
    
    public static void hbaseSimplePutData(HTable table, String rowKey, HashSet<Integer> vals) {
        Put put = null;
        try {
            int i = 3;
            byte[] hbaseValue = new byte[vals.size() * 4];
            for (int val : vals) {
                for (int n = 0; n < 4; n++) {
                    hbaseValue[i - n] = (byte) ((val >> (8 * n)) & 0xFF);
                }
                i += 4;
            }
            put = new Put(Bytes.toBytes(rowKey));
            put.add("value".getBytes(), "".getBytes(), hbaseValue);
            table.put(put);
        } catch (Exception ex) {
            log.error(ex, ex);
        }
    }

    public static void hbasePutData(String tableName, String rowKey, String defaultFamily, ArrayList<String> columns, Map<Utf8,Object> recordMap) {
        HTable table = null;
       
        try {
            if (GlobalParas.connection == null) {
                initHbaseConnetion();
            }
//            try {                
//                HBaseAdmin admin = new HBaseAdmin(GlobalParas.conf);             
//                String[] fam = null;
//                fam[0] = defaultFamily;               
//                if (!admin.tableExists(tableName)) {
//                    log.info(tableName + " don't exit,will create it");
//                    hbaseCreateTable(tableName, fam, 7);
//                }
//            } catch (Exception ex) {
//                log.info("admin false:"+ex);
//            }              
            table = new HTable(GlobalParas.conf, Bytes.toBytes(tableName));            
            table.setAutoFlush(false);
            table.setOperationTimeout(15000);
            table.setWriteBufferSize(1024 * 1024 * 2);
        } catch (Exception e) {
            log.info("create the table failed!");
        }
        try {
            Put put = new Put(Bytes.toBytes(rowKey));
            //Put put = new Put(Bytes.toBytes(DigestUtils.md5Hex(rowKey)));
            //Put put = new Put(Bytes.toBytes(rowKey));
            for (String qulifier : columns) {
                Utf8 qulifier1= new Utf8(qulifier);
                if (recordMap.get(qulifier1) == null) {
                    continue;
                }
                put.add(Bytes.toBytes(defaultFamily),
                        Bytes.toBytes(qulifier), Bytes.toBytes(recordMap.get(qulifier1).toString()));
            }            
            table.put(put);            
        } catch (Exception ex) {
            log.error(ex, ex);
        }
        try {
            table.flushCommits();
            table.close();
        } catch (Exception e) {
            log.error(e, e);
        }
    }

    public static void hbaseBatchPutData(String tableName, String defaultFamily, ArrayList<String> columns, ArrayList<HashMap<String, GenericRecord>> val) {
        HTable table = null;
        List<Put> list = new ArrayList<Put>();

        try {
            table = new HTable(GlobalParas.conf, Bytes.toBytes(tableName));
            table.setAutoFlush(false);
            table.setOperationTimeout(15000);
            table.setWriteBufferSize(1024 * 1024 * 2);
        } catch (Exception e) {
            log.error(e, e);
        }
        try {
            Iterator<String> itor = null;
            String rowkey = null;
            for (HashMap<String, GenericRecord> rowkey2Record : val) {
                itor = rowkey2Record.keySet().iterator();
                if (itor.hasNext()) {
                    rowkey = itor.next();
                }
                GenericRecord record = rowkey2Record.get(rowkey);
                Put put = new Put(Bytes.toBytes(rowkey));
                for (String qulifier : columns) {
                    if (record.get(qulifier) == null) {
                        continue;
                    }
                    put.add(Bytes.toBytes(defaultFamily),
                            Bytes.toBytes(qulifier), Bytes.toBytes(record.get(qulifier).toString()));
                }
                list.add(put);
            }
            table.put(list);
            log.info("put data to hbase successfully");
        } catch (Exception ex) {
            log.error(ex, ex);
        }
        try {
            table.flushCommits();
            table.close();
            list.clear();
        } catch (Exception e) {
            log.error(e, e);
        }
    }

    public static boolean hbaseCreateTable(String tableName, String[] columnFamily, int maxVersion) {
        try {
            HBaseAdmin admin = new HBaseAdmin(GlobalParas.conf);
            if (admin.tableExists(tableName)) {
                log.warn(tableName + " is exist,please to delete it first....");
            } else {
                HTableDescriptor tableDes = new HTableDescriptor(tableName);
                for (int i = 0; i < columnFamily.length; i++) {
                    HColumnDescriptor colDes = new HColumnDescriptor(columnFamily[i]);
                    colDes.setMaxVersions(maxVersion);
                    tableDes.addFamily(colDes);
                }
                admin.createTable(tableDes);
            }
            admin.close();
        } catch (MasterNotRunningException e) {
            log.error(e, e);
            return false;
        } catch (ZooKeeperConnectionException e) {
            log.error(e, e);
            return false;
        } catch (IOException e) {
            log.error(e, e);
            return false;
        } catch (Exception e) {
            log.error(e, e);
            return false;
        }
        return true;
    }

    public static void hbaseFlush(HTable table) {
        try {
            table.flushCommits();
        } catch (Exception ex) {
            log.error(ex, ex);
        }
    }

    public static void hbaseCloseTable(HTable table) {
        try {
            table.flushCommits();
            table.close();
        } catch (Exception e) {
            log.error(e, e);
        }
    }

    public static boolean hbaseDeleteTable(String tableName) {
        try {
            HBaseAdmin admin = new HBaseAdmin(HbaseUtilOp.conf);
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            admin.close();
        } catch (MasterNotRunningException e) {
            log.error(e, e);
        } catch (ZooKeeperConnectionException e) {
            log.error(e, e);
        } catch (IOException e) {
            log.error(e, e);
        } catch (Exception e) {
            log.error(e, e);
        }
        return true;
    }

    public static void scanAllData(String tableName, int maxVersoin) {
        HTable table = null;
        ResultScanner rs = null;
        try {
            Scan scan = new Scan();
            table = new HTable(GlobalParas.conf, tableName);
            scan.setMaxVersions(maxVersoin);
            rs = table.getScanner(scan);
            log.info(rs);
            for (Result r : rs) {
                for (KeyValue kv : r.list()) {
                    log.info("row:" + Bytes.toString(kv.getRow()));
                    log.info("family:" + Bytes.toString(kv.getFamily()));
                    log.info("qualifier:" + Bytes.toString(kv.getQualifier()));
                    log.info("value:" + Bytes.toString(kv.getValue()));
                    log.info("timestamp:" + kv.getTimestamp());
                }
            }

        } catch (Exception e) {
            log.error(e, e);
        } finally {
            try {
                rs.close();
                table.close();
            } catch (IOException ex) {
                log.error(ex, ex);
            }
        }
    }

//    public static void main(String[] args) {
//        PropertyConfigurator.configure("log4j.properties");
//        HbaseUtilOp.initHbaseConnetion("192.168.120.221,192.168.120.222", 2187, "192.168.120.222:60013");
//        HbaseUtilOp.hbaseDeleteTable("test_table");
//        String[] fami = new String[2];
//        fami[0] = "value";
//        fami[1] = "field2";
//        HbaseUtilOp.hbaseCreateTable("test_table", fami, 7);
//
//        HTable table = HbaseUtilOp.getDefaultHTable("192.168.120.221,192.168.120.222", 2187, "test_table", "192.168.120.222:60013");
//        System.out.println(table);
//        HashSet<Integer> set = new HashSet<Integer>();
//        set.add(10);
//        set.add(11);
//        HbaseUtilOp.hbaseSimplePutData(table, "test_table", set);
//        HbaseUtilOp.hbaseFlush(table);
//        HbaseUtilOp.scanAllData("test_table", 10);
//    }
}
